package org.jsets.fastboot.concurrent;

import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.jsets.fastboot.util.SpringContextHolder;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * 异步执行器 <br>
 * 线程数默认为当前CPU核心数据的2倍 <br>
 * 采用静态内部类的单例模式 <br>
 * 外部类加载时并不需要立即加载内部类， <br>
 * 内部类不被加载则不去初始化INSTANCE， <br>
 * 故如果不使用则不占内存<br><br>
 * 
 * 线程池的拒绝策略为调用者执行<br>
 * 因此，队列没有满的时候是异步的，满的时候就阻塞了<br>
 * 这样不会丢任务，但是请留意主线程的压力
 * 
 * @author wangjie (https://github.com/wj596)
 * @date 2022.01.11 23:56
 */
@Slf4j
public class AsyncActuator {

	private static class SingleHolder {
		private static final AsyncActuator INSTANCE = new AsyncActuator();
	}
	
	public static AsyncActuator getInstance() {
		return SingleHolder.INSTANCE;
	}
	
	private final ThreadPoolExecutor executor;

	private AsyncActuator() {
		Integer nThreads = SpringContextHolder.getProperty("async-actuator.nthreads", Integer.class);
		if(Objects.isNull(nThreads)) {
			nThreads  = Runtime.getRuntime().availableProcessors() * 2;
		}
		log.info("create AsyncActuator ThreadPool nThreads is[{}]", nThreads);
		this.executor = new ThreadPoolExecutor(nThreads,
				nThreads,
				0L,
				TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue<Runnable>(),
				new NamedThreadFactory("async_actuator"),
				new RejectedExecutionHandler() {
					// 拒绝策略为调用者执行
					// 因此，队列没有满的时候是异步的，满的时候就阻塞了
					@Override
					public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
						log.warn("AsyncActuator workQueue is full, note the main thread pressure!!");
						r.run();
					}
	            });
	}
	
	
	public static void monitoring() {
		int corePoolSize = getInstance().executor.getCorePoolSize();
		int maximumPoolSize = getInstance().executor.getMaximumPoolSize();
		int largestPoolSize = getInstance().executor.getLargestPoolSize();
		int activeCount = getInstance().executor.getActiveCount();
		long taskCount = getInstance().executor.getTaskCount();
		long completedTaskCount = getInstance().executor.getCompletedTaskCount();
		int queueSize = getInstance().executor.getQueue().size();
		StringBuilder sb = new StringBuilder();
		sb.append("AsyncActuator Snapshoot: ");
		sb.append("\r\n corePoolSize: "+corePoolSize);
		sb.append("\r\n maximumPoolSize: "+maximumPoolSize);
		sb.append("\r\n largestPoolSize: "+largestPoolSize);
		sb.append("\r\n activeCount: "+activeCount);
		sb.append("\r\n taskCount: "+taskCount);
		sb.append("\r\n completedTaskCount: "+completedTaskCount);
		sb.append("\r\n queueSize: "+queueSize);
		log.info(sb.toString());
	}
	
	public static ExecutorService getDefaultExecutor() {
		return getInstance().executor;
	}

	public static void execute(final Runnable runnable) {
		getDefaultExecutor().execute(runnable);
	}

	public static <T> Future<T> submit(Callable<T> task) {
		return getDefaultExecutor().submit(task);
	}

	public static <T> Future<T> submit(Runnable task, T result) {
		return getDefaultExecutor().submit(task, result);
	}
	
	public static CompletableFuture<Void> runAsync(Runnable task) {
		return CompletableFuture.runAsync(task, getDefaultExecutor());
	}
	
	public static <T> CompletableFuture<T> supplyAsync(Supplier<T> task) {
		return CompletableFuture.supplyAsync(task, getDefaultExecutor());
	}
	
	public static void shutdown() {
		if(Objects.nonNull(getDefaultExecutor())&&!getDefaultExecutor().isShutdown()) {
			log.info("AsyncActuator ThreadPool shutdown");
			getDefaultExecutor().shutdown();
		}
	}
}